package com.liufei.producer;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.HashMap;
import java.util.Map;


/**
 * @author 刘飞
 * @time 2021/4/20
 */
@Service
@Slf4j
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;

    public static final String TOPIC_ONE = "topic.gps";
    public static final String TOPIC_GROUP1 = "topic.group1";
    public static final String TOPIC_GROUP2 = "topic.group2";

    public void send(Object object){
        String objString = JSONObject.toJSONString(object);
        log.info("准备发送消息：{}",objString);
        Map map = new HashMap<String, String>();
        map.put("name","刘飞");
        ListenableFuture<SendResult<String,Object>>  future= kafkaTemplate.send(TOPIC_ONE,objString);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.info("消息发送失败");
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                log.info("消息发送成功");
            }
        });
    }
}
